package net.cassandraview.db;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.KsDef;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Class containing methods to create connections to database.
 * <p/>
 * User: nagakhl
 * Date: 21.01.12
 * Time: 21:37
 */
public class ConnectionManager {

    protected Logger log = Logger.getLogger(getClass().getName());

    private ConnectionPool<TProtocol> pool;

    private final DbManagerConfiguration config;

    private final ConnectionProducer<TProtocol> producer;

    private final Map<String, CfDef> schemaCache = new HashMap<String, CfDef>();

    private final ReentrantLock lock = new ReentrantLock();

    private final Condition updating;

    private boolean isUpdating = false;

    /**
     * Create database connection. Connection will be created, not taken from pool.
     * New socket will be opened each time when connection is created. And it's up to client code to close this
     * connection by calling ((ThriftConnection) connection).close()
     * <p/>
     * Also no default keyspace will be set for this connection.
     *
     * @param host
     * @param port
     * @return
     */
    public static Connection createConnection(String host, int port) {
//        TTransport transport = new TFastFramedTransport(new TSocket(host, port));
//        TProtocol protocol = new TBinaryProtocol(transport);
        throw new UnsupportedOperationException("Not implemented yet.");
    }

    public ConnectionManager(DbManagerConfiguration configuration) {
        this.config = configuration;
        this.updating = lock.newCondition();
        this.producer = new ConnectionProducer<TProtocol>() {
            @Override
            public TProtocol initalizeConnection() {
                try {
                    TTransport transport = new TFastFramedTransport(new TSocket(config.getHost(), config.getPort()));
                    TProtocol protocol = new TBinaryProtocol(transport);
                    transport.open();
                    return protocol;
                } catch (TTransportException e) {
                    throw new ConnectionInitializationException("Cannot open connection to database.", e);
                }
            }
        };
        pool = new ConnectionPool<TProtocol>(configuration.getInitialConnections(), configuration.getMaxConnections(), producer);
    }

    public Connection getConnection() {
        return getManagementConnection();
    }

    public ManagementConnection getManagementConnection() {
        TProtocol proto = pool.getConnection();
        ThriftManagementConnection c = new ThriftManagementConnection();
        c.setConsistencyLevel(config.getReadConsistency());
        c.setLimit(config.getDefaultLimit());
        c.setKeyspace(config.getDefaultKeyspace());
        Cassandra.Client client = new Cassandra.Client(proto);
        try {
            client.set_keyspace(config.getDefaultKeyspace());
        } catch (Exception e) {
            log.log(Level.SEVERE, "Cannot set client default keyspace.", e);
        }
        c.setClient(client);
        c.setParentManager(this);
        return c;
    }

    void finish(TProtocol protocol) {
        pool.returnConnection(protocol);
    }

    void updateSchema() {
        lock.lock();
        try {
            isUpdating = true;
            schemaCache.clear();
            Cassandra.Client client = new Cassandra.Client(producer.initalizeConnection());
            KsDef ksDef = client.describe_keyspace(config.getDefaultKeyspace());
            List<CfDef> cfDefs = ksDef.getCf_defs();
            for (CfDef cfd : cfDefs) {
                schemaCache.put(cfd.getName(), cfd);
            }
            client.getInputProtocol().getTransport().close();
            isUpdating = false;
            updating.notifyAll();
        } catch (Exception e) {
            log.log(Level.SEVERE, "Error updating database schema.", e);
        } finally {
            lock.unlock();
        }
    }

    void addToSchema(CfDef cfDef) {
        lock.lock();
        try {
            isUpdating = true;
            schemaCache.put(cfDef.getName(), cfDef);
            isUpdating = false;
            updating.signalAll();
        } catch (Exception e) {
            log.log(Level.SEVERE, "Error while trying to add ColumnFamily to schema definition.", e);
        } finally {
            lock.unlock();
        }
    }

    CfDef getCfDescription(String name) {
        try {
            while (isUpdating) {
                updating.await();
            }
            if (!schemaCache.containsKey(name)) {
                updateSchema(); //if schema changed from outside of this connection manager
            }
            return schemaCache.get(name);
        } catch (InterruptedException e) {
            log.log(Level.SEVERE, "Interrupted while waiting for update of schema.", e);
        }
        return null;
    }

}
